package site.weiyikai.eventdriven.core;


import lombok.Data;

import java.util.List;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.function.Predicate;
import java.util.stream.Collectors;


/**
 * 事件处理器实现类
 * 实现了事件发布服务，具有接收事件的功能；
 * 依赖了事件交付器，用于将事件交付给交付器；
 * 实现了事件订阅服务，维护了事件和事件订阅器之间的关系，相当于一个事件订阅器。
 * 业务服务可以自定义自己需要的事件发布器（依赖或继承事件处理器），然后在发布器中生成事件，调用事件处理器的publish()方法，发布事件，当然也可以直接调用事件处理器的发布方法。
 *
 * @author 程序员小魏
 * 创建时间: 2023/11/07
 */
@Data
public class EventServiceImpl implements EventService {

    private Delivery delivery;
    private final ConcurrentMap<SubscriptionKey, EssProvider> essProviders = new ConcurrentHashMap<>();
    private static final String DEFAULT_CHANNEL = "DEFAULT_CHANNEL";


    public EventServiceImpl(Delivery delivery) {
        this.delivery = delivery;
    }

    @Override
    public <T> void publish(T event) {
        this.publish(DEFAULT_CHANNEL, event);
    }

    @Override
    public <T> void publish(String channel, T event) {
        //事件分配
        List<EventSubscriber<T>> eventSubscribers = (List) this.essProviders.entrySet()
                .stream()
                .filter(entry -> (entry.getKey()).matches(channel, event))
                .flatMap(entry -> (entry.getValue()).get().getEventSubscribers().stream())
                .collect(Collectors.toList());
        //分配完，交付给事件交付器
        this.delivery.publishTo(eventSubscribers, event);
    }

    @Override
    public void subscribe(Predicate<?> eventFilter, EventSubscriber<?> subscriber) {
        this.subscribe(DEFAULT_CHANNEL, eventFilter, subscriber);
    }

    @Override
    public void subscribe(String channel, Predicate<?> eventFilter, EventSubscriber<?> subscriber) {
        SubscriptionKey key = new SubscriptionKey(channel, eventFilter);
        EssProvider newProvider = new EssProvider();
        EssProvider essProvider = this.essProviders.putIfAbsent(key, newProvider);
        if (essProvider == null) {
            essProvider = newProvider;
        }
        EventSubscriberSet subscribers = essProvider.get();
        subscribers.add(subscriber);
    }

    @Override
    public void unsubscribe(EventSubscriber<?> subscriber) {
        this.essProviders.values().forEach(essProvider -> essProvider.get().remove(subscriber));
    }

    @Override
    public void unsubscribe(String channel, Predicate<?> eventFilter, EventSubscriber<?> subscriber) {
        SubscriptionKey key = new SubscriptionKey(channel, eventFilter);
        Optional.ofNullable(this.essProviders.get(key)).ifPresent(essProvider -> essProvider.get().remove(subscriber));
    }


    private static class EssProvider {
        private EventSubscriberSet ess;
        private EssProvider() {
        }
        public synchronized EventSubscriberSet get() {
            if (this.ess == null) {
                this.ess = new EventSubscriberSet();
            }
            return this.ess;
        }
    }
}
